package priv.pfz.raft.peer;

import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * raft事件的延迟队列，所有raft异步操作均作为事件提交（add）到此队列，到时间自动触发执行
 * 保证线程安全：队列管理和事件执行是串行的，清除队列（clear）时，已取出的事件也不会再执行
 * 提供暂停（pause）和恢复（resume）功能
 * @author pengfangzhou
 * @date 2022/2/6 19:00
 */
@Slf4j
public class RaftEventQueue implements Runnable {
    private final DelayQueue<RaftEvent> queue = new DelayQueue<>();
    private boolean pausing = false;
    private final List<RaftEvent> freezeList = Lists.newArrayList();
    private final ExecutorService executorService = Executors.newSingleThreadExecutor();
    private volatile int version = 0;
    private final Object lock;
    private static final Joiner TO_STR_JOINER = Joiner.on(",");

    public RaftEventQueue() {
        executorService.submit(this);
        this.lock = new Object();
    }

    public RaftEventQueue(Object lock) {
        executorService.submit(this);
        this.lock = lock;
    }

    @Override
    public void run() {
        while (true) {
            try {
                RaftEvent raftEvent = queue.take();
                if (raftEvent.getVersion() != version) {
                    continue;
                }
                synchronized (lock) {
                    if (raftEvent.getVersion() != version) {
                        continue;
                    }
                    raftEvent.getRunnable().run();
                }
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    /**
     * 添加事件
     * @param name 事件名称
     * @param runnable 事件触发执行的操作
     * @param delay  delay毫秒之后触发
     */
    public void add(String name, Runnable runnable, long delay) {
        long effectTime = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(delay);
        queue.add(new RaftEvent(name, runnable, effectTime, version));
    }

    /**
     * 移除队列中所有事件
     */
    public synchronized void clear() {
        synchronized (lock) {
            queue.clear();
            version++;
        }
    }

    /**
     * 暂停时间流逝
     */
    public void pause() {
        synchronized (lock) {
            long now = System.nanoTime();
            for (RaftEvent raftEvent : queue) {
                raftEvent.setLeftTime(raftEvent.getEffectTime() - now);
                raftEvent.setPausing(true);
                freezeList.add(raftEvent);
            }
            queue.clear();
            pausing = true;
        }
    }

    /**
     * 恢复时间流逝
     */
    public void resume() {
        long now = System.nanoTime();
        for (RaftEvent raftEvent : freezeList) {
            queue.add(new RaftEvent(
                    raftEvent.getName(),
                    raftEvent.getRunnable(),
                    now + raftEvent.getLeftTime(),
                    raftEvent.getVersion()));
        }
        freezeList.clear();
        pausing = false;
    }

    @Override
    public String toString() {
        return TO_STR_JOINER.join(pausing ? freezeList : queue);
    }
}
